package com.hm.kafkademo.consumer;

import com.alibaba.fastjson2.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.hm.kafkademo.dto.Camera;
import com.hm.kafkademo.entity.AnalysisResult;
import com.hm.kafkademo.entity.HumanSnap;
import com.hm.kafkademo.entity.Rect;
import com.hm.kafkademo.entity.Target;
import com.hm.kafkademo.mapper.CameraMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;

@Slf4j
@Component
@RequiredArgsConstructor
public class HmConsumer2 {

    @Autowired
    private CameraMapper cameraMapper;

    @Resource
    private KafkaTemplate<String, String> firstkafkaTemplate;

    @KafkaListener(
            containerFactory = "secondKafkaListenerContainerFactory",
            topics = {"HIK_SMART_METADATA_TOPIC"},
            groupId = "testGroup12")
    public void consumer(ConsumerRecord<String, String> record) throws JsonProcessingException {
        // 获取数据库

        //获取报文
        log.info("------HIK_SMART_METADATA_TOPIC报文信息-------：{} ......", JSON.toJSONString(record.value()));
        // 把报文转换成实体类
        HumanSnap humanSnap = JSON.parseObject(record.value(), HumanSnap.class);
        // 获取rect信息
        List<AnalysisResult> analysisResult = humanSnap.getAnalysisResult();
        List<Target> target = analysisResult.get(0).getTarget();
        Rect rect = target.get(0).getRect();
        // 判断是否是0
        if (rect.getHeight() != 0.0 && rect.getWidth() != 0.0 && rect.getX() != 0.0 && rect.getY() != 0.0) {
            // 插入新的kafka
            log.info("-----有坐标信息，开始查询数据库。");
            // 获取相机内码
            String cameraIndexCode = analysisResult.get(0).getTargetAttrs().getCameraIndexCode();
            LambdaQueryWrapper<Camera> cameraLambdaQueryWrapper = new LambdaQueryWrapper<>();
            cameraLambdaQueryWrapper.eq(Camera::getIndexCode, cameraIndexCode);
            List<Camera> cameras = cameraMapper.selectList(cameraLambdaQueryWrapper);
            if (!cameras.isEmpty()) {
                log.info("-----有相机信息，开始传输数据。{}",cameras.get(0).getExternalCode());
                // 把相机的外码赋值给相机内码字段
                humanSnap.getAnalysisResult().get(0).getTargetAttrs().setCameraIndexCode(cameras.get(0).getExternalCode());
                // 把数据写入kafka
                firstkafkaTemplate.send("test", JSON.toJSONString(humanSnap));
            } else {
                log.info("-----无相机信息，数据丢弃{}", cameraIndexCode);
            }
        } else {
            log.info("-----无坐标信息，数据丢弃。");
        }


    }
}
